package org.apache.ocean.main.name;

import java.io.File;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.lang.StringUtils;
import org.apache.ocean.FieldUtil;
import org.apache.ocean.main.database.JDBCDatabase;
import org.apache.ocean.main.database.SQLDefaultParameters;
import org.apache.ocean.main.database.SQLParameters;
import org.apache.ocean.main.database.SQLResults;
import org.apache.ocean.main.database.SQLRow;

public class SQLNameDatabase extends JDBCDatabase {
	public static NameDefaults NAME_DEFAULTS = new NameDefaults();
	private AtomicLong replicationIdSequence;

	public SQLNameDatabase(File directoryFile) throws Exception {
		super("namedatabase", directoryFile);
		Long highestReplicationId = findHighestReplicationId();
		if (highestReplicationId == null) {
			replicationIdSequence = new AtomicLong(0);
		} else {
			replicationIdSequence = new AtomicLong(highestReplicationId);
		}
		createTable("node", Node.class);
		syncColumns("node");
		createTable("cell", Cell.class);
		syncColumns("cell");
	}
  
	public List<Node> findNodesByCell(String cell) throws Exception {
		String sql = "SELECT * FROM node WHERE reportedCellId >= ?";
		SQLResults sqlResults = executeQuery(sql, new SQLParameters(cell, Types.VARCHAR), "node");
		return sqlResults.toList(new NodeTransform());
	}
	
	public List<Node> findNodes(long replicationId) throws Exception {
		String sql = "SELECT * FROM node WHERE replicationid >= ?";
		SQLResults sqlResults = executeQuery(sql, new SQLParameters(replicationId, Types.BIGINT), "node");
		return sqlResults.toList(new NodeTransform());
	}

	public List<Cell> findCells(long replicationId) throws Exception {
		String sql = "SELECT * FROM cell WHERE replicationid >= ?";
		SQLResults sqlResults = executeQuery(sql, new SQLParameters(replicationId, Types.BIGINT), "node");
		return sqlResults.toList(new CellTransform());
	}

	public class CellTransform implements SQLResults.Transform<Cell> {
		public Cell row(SQLRow row) throws Exception {
			return toCell(row);
		}
	}

	public Cell toCell(SQLRow row) throws Exception {
		return (Cell) rowToObject(row, Cell.class, NAME_DEFAULTS);
	}

	public class NodeTransform implements SQLResults.Transform<Node> {
		public Node row(SQLRow row) throws Exception {
			return toNode(row);
		}
	}

	public Node toNode(SQLRow row) throws Exception {
		return (Node) rowToObject(row, Node.class, NAME_DEFAULTS);
	}

	public Long findHighestReplicationId() throws Exception {
		List<Long> list = new ArrayList<Long>();
		Long nodeValue = findHighestReplicationId("node");
		if (nodeValue != null)
			list.add(nodeValue);
		Long cellValue = findHighestReplicationId("cell");
		if (cellValue != null)
			list.add(cellValue);
		return Collections.max(list);
	}

	public Long findHighestReplicationId(String tableName) throws Exception {
		String sql = "SELECT MAX(replicationId) FROM " + tableName;
		return executeQuery(sql, tableName).getAggregateLong();
	}

	public Node updateNode(Node node) throws Exception {
		return (Node) updateRow("node", node, NAME_DEFAULTS);
	}
	
	public Cell updateCell(Cell cell) throws Exception {
		return (Cell) updateRow("cell", cell, NAME_DEFAULTS);
	}

	public Set<String> findDistinctCellIDs() throws Exception {
		String sql = "SELECT DISTINCT(assignedCellID) FROM node";
		SQLResults results = executeQuery(sql, null, "node");
		Set<String> cellIDs = new HashSet<String>(results.size());
		for (SQLRow row : results) {
			if (row.size() == 0)
				throw new Exception("row size 0: " + row);
			String cellID = (String) row.get(0);
			if (cellID != null) {
				cellIDs.add(cellID);
			}
		}
		return cellIDs;
	}
  
	public void insertCell(Cell cell) throws Exception {
		cell.replicationId = replicationIdSequence.incrementAndGet();
		cell.dateCreated = new Date();
		insert("cell", cell, null);
	}

	public void updateCellNodes(String cellID, Collection<Node> nodes) throws Exception {
		// TODO: handle setting replicationid sequence
		for (Node node : nodes) {
			node.replicationId = replicationIdSequence.incrementAndGet();
		}
		for (Node node : nodes) {
			updateNode(node);
		}
	}

	public static class NameDefaults extends SQLDefaultParameters {
		public void setValues(String action, Object object) throws Exception {
			if (object instanceof Cell) {
				FieldUtil.setFieldValue(object, "dateCreated", new Date());
			}
			if (object instanceof Node) {
				if (StringUtils.equals("update", action)) {
					FieldUtil.setFieldValue(object, "dateModified", new java.util.Date());
				} else if (StringUtils.equals("insert", action)) {
					java.util.Date date = new java.util.Date();
					FieldUtil.setFieldValue(object, "dateModified", date);
					FieldUtil.setFieldValue(object, "dateCreated", date);
				}
			}
		}
	}
}
